package com.apps.sdses.flink141.demo;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;

/**
 * @author jiwei
 * @description
 * @date 2023/9/15 14:00
 */
public class FlinkWindowDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<Tuple3<String, Integer, Long>> dataStream = env.fromElements(
                Tuple3.of("SensorA", 10, 1694742900000L),   //2023-09-15 09:55:00
                Tuple3.of("SensorA", 11, 1694743200000L),   //2023-09-15 10:00:00
                Tuple3.of("SensorA", 12, 1694743202000L),   //2023-09-15 10:00:02
                Tuple3.of("SensorA", 13, 1694743204000L),   //2023-09-15 10:00:04
                Tuple3.of("SensorA", 14, 1694743206000L),   //2023-09-15 10:00:06
                Tuple3.of("SensorA", 15, 1694743208000L),   //2023-09-15 10:00:08
                Tuple3.of("SensorA", 16, 1694743210000L),   //2023-09-15 10:00:10
                Tuple3.of("SensorA", 17, 1694743212000L),   //2023-09-15 10:00:12

                Tuple3.of("SensorB", 21, 1694743140000L),   //2023-09-15 09:59:00
                Tuple3.of("SensorB", 22, 1694743200000L),   //2023-09-15 10:00:00
                Tuple3.of("SensorB", 23, 1694743202000L),   //2023-09-15 10:00:02
                Tuple3.of("SensorB", 24, 1694743204000L),   //2023-09-15 10:00:04
                Tuple3.of("SensorB", 25, 1694743206000L),   //2023-09-15 10:00:06
                Tuple3.of("SensorB", 26, 1694743208000L),   //2023-09-15 10:00:08
                Tuple3.of("SensorB", 27, 1694743215000L)    //2023-09-15 10:00:15
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, Integer, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((event, timestamp) -> event.f2));

        //按传感器名 分组后 开窗
//        dataStream
//                .keyBy(event -> event.f0)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<Tuple3<String, Integer, Long>, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String s, ProcessWindowFunction<Tuple3<String, Integer, Long>, String, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, Long>> iterable, Collector<String> collector) throws Exception {
//                        SimpleDateFormat outputFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                        String start = outputFormat.format(new Date(context.window().getStart()));
//                        String end = outputFormat.format(new Date(context.window().getEnd()));
////                        long start = context.window().getStart();
////                        long end = context.window().getEnd();
//                        LinkedList<String> data = new LinkedList<>();
//                        for (Tuple3<String, Integer, Long> event : iterable) {
//                            data.add(event.toString());
//                        }
//                        collector.collect("窗口开始时间:" + start + ", 窗口结束时间: " + end  + ", 窗口内有: "+ data.size() + " 条数据, 数据为: " + data);
//                    }
//                })
//                .print();

        //不分组后开窗
        dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                        .process(new ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, Long>> iterable, Collector<String> collector) throws Exception {
                                SimpleDateFormat outputFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                String start = outputFormat.format(new Date(context.window().getStart()));
                                String end = outputFormat.format(new Date(context.window().getEnd()));
                                LinkedList<String> data = new LinkedList<>();
                                for (Tuple3<String, Integer, Long> event : iterable) {
                                    data.add(event.toString());
                                }
                                collector.collect("窗口开始时间:" + start + ", 窗口结束时间: " + end  + ", 窗口内有: "+ data.size() + " 条数据, 数据为: " + data);

                            }
                        })
                        .print();











        //滑动事件时间窗口
//        dataStream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
//                .process(new ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>() {
//                    @Override
//                    public void process(ProcessAllWindowFunction<Tuple3<String, Integer, Long>, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, Long>> iterable, Collector<String> collector) throws Exception {
//                        SimpleDateFormat outputFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                        String start = outputFormat.format(new Date(context.window().getStart()));
//                        String end = outputFormat.format(new Date(context.window().getEnd()));
//                        LinkedList<String> data = new LinkedList<>();
//
//                        for (Tuple3<String, Integer, Long> event : iterable) {
//                            data.add(event.toString());
//                        }
//                        collector.collect("窗口开始时间:" + start + ", 窗口结束时间: " + end  + ", 窗口内有: "+ data.size() + " 条数据, 数据为: " + data);
//                    }
//                })
//                .print();


        env.execute();
    }
}
